/**
 * ProducerServiceImpl.java
 *
 * @author: ZhuJiahui
 * @date: 2019/1/9 22:35
 */
package com.zhujiahui.sbkafka.second;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

/**
 * @description:
 * @author: ZhuJiahui
 * @version: 1.0
 */
@Service
public class ProducerServiceImpl {
    private static final Logger LOGGER = LoggerFactory.getLogger(ProducerServiceImpl.class);

    @Autowired
    private KafkaTemplate kafkaTemplate;

    //发送消息方法
    public void sendJson(String topic, String message) {
        JSONObject jsonObj = new JSONObject();

        jsonObj.put("message", message);
        jsonObj.put("topic", topic);
        jsonObj.put("ts", System.currentTimeMillis() + "");

        LOGGER.info("json+++++++++++++++++++++  message = {}", jsonObj.toJSONString());

        ListenableFuture<SendResult<String, String>>
                future = kafkaTemplate.send(topic, jsonObj.toJSONString());
        future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
            @Override
            public void onSuccess(SendResult<String, String> result) {
                System.out.println("msg OK." + result.toString());
            }

            @Override
            public void onFailure(Throwable ex) {
                System.out.println("msg send failed: ");
            }
        });
    }
}
